package com.mjf.networkflow_analysis

import java.lang

import com.mjf.dim.{UserBehavior, UvCount}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.{Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import redis.clients.jedis.Jedis

/**
 * UV 布隆过滤器实现
 *
 * 一亿用户去重计算uv
 * 假设一个用户UserId占用100B，一共需要占用 10`^`8 * 100B = 10GB
 * 如果可以用1bite代替用户是否存在，一共需要占用 10`^`8 * 1bite ~ 10MB
 * 为了降低hash碰撞，可以在10MB的基础上扩充N倍
 */
object UvWithBloomFilter {
  def main(args: Array[String]): Unit = {

    // 创建流处理执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    // 定义时间语义
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 从文件读取数据
    val inputStream: DataStream[String] = env.readTextFile("D:\\coding\\idea\\UserBehaviorAnalysis\\HotItemsAnalysis\\src\\main\\resources\\UserBehavior.csv")

    // 将数据转换为样例类，并且提取timestamp定义watermark
    val dataStream: DataStream[UserBehavior] = inputStream.map {
      line =>
        val dataArray: Array[String] = line.split(",")
        UserBehavior(dataArray(0).toLong, dataArray(1).toLong, dataArray(2).toInt, dataArray(3), dataArray(4).toLong)
    }.assignAscendingTimestamps(_.timestamp * 1000L)

    // 分配key,包装成二元组开窗聚合
    val uvStream: DataStream[UvCount] = dataStream
      .filter(_.behavior == "pv")
      .map(data => ("uv", data.userId))
      .keyBy(_._1)
      .timeWindow(Time.hours(1))
      .trigger(new MyTrigger()) // 自定义Trigger
      .process(new UvCountResultWithBloomFilter())

    uvStream.print()

    env.execute("uv job")

  }
}

// 自定义触发器，每来一条数据就触发一次窗口计算
class MyTrigger extends Trigger[(String, Long), TimeWindow] {
  // 数据来了之后，触发计算并清空状态，不保存数据
  override def onElement(element: (String, Long), timestamp: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = TriggerResult.FIRE_AND_PURGE

  override def onProcessingTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = TriggerResult.CONTINUE

  override def onEventTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = TriggerResult.CONTINUE

  override def clear(window: TimeWindow, ctx: Trigger.TriggerContext): Unit = {}
}

// 自定义ProcessWindowFunction,把当前数据进行处理,位图保存在redis中
class UvCountResultWithBloomFilter() extends ProcessWindowFunction[(String, Long), UvCount, String, TimeWindow] {

  var jedis: Jedis = _
  var bloom: Bloom = _

  override def open(parameters: Configuration): Unit = {
    jedis = new Jedis("hadoop103", 6379)
    // bitmap大小10亿bite:也就是2^30,占用128M
    bloom = new Bloom(1 << 30)
  }

  // 每来一个数据，主要是要用布隆过滤器判断redis位图中对应位置是否为1
  override def process(key: String, context: Context, elements: Iterable[(String, Long)], out: Collector[UvCount]): Unit = {
    // bitmap用当前窗口的end作为key,保存到redis中（windowEnd，bitmap）
    val storedKey: String = context.window.getEnd.toString

    // 把每个窗口的uv count值，作为状态也存入redis中，存成一张叫做countMap的表
    val countMap: String = "countMap"
    // 先获取当前的count值
    var count: Long = 0L

    if (jedis.hget(countMap, storedKey) != null) {
      count = jedis.hget(countMap, storedKey).toLong
    }

    // 取userId,计算hash值,判断是否在bitmap中
    val userId: String = elements.head._2.toString
    val offset: Long = bloom.hash(userId, 61)
    val isExist: lang.Boolean = jedis.getbit(storedKey, offset)

    // 如果不存在，那么就将对应位置置为1，count加1；如果存在，不做操作
    if (!isExist) {
      jedis.setbit(storedKey, offset, true)
      jedis.hset(countMap, storedKey, (count + 1).toString)
    }
  }

}

// 自定义一个布隆过滤器
class Bloom(size: Long) extends Serializable {
  // 定义位图的大小,应该是2的整次幂
  private val cap = size

  // 实现一个hash函数（hash函数设计不完善会导致hash碰撞高）
  def hash(str: String, seed: Int): Long = {
    var result: Int = 0
    for (i <- 0 until str.length) {
      result = result * seed + str.charAt(i)
    }

    // 返回一个在cap范围内的一个值
    (cap - 1) & result  // 位运算中1&1=1,1&0=0,0&0=0
  }
}









